package com.shujia.flink.core;

import com.shujia.flink.bean.Alert;
import com.shujia.flink.bean.Transaction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

public class Demo9TransactionProcess {
    public static void main(String[] args) throws Exception {
        /*
         * 001,100
         * 001,200
         * 001,0.01
         * 001,600
         * 001,200
         * 001,0.2
         * 001,300
         * 001,700
         */

        /*
         * 对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //解析数据
        DataStream<Transaction> transDS = linesDS.map(line -> {
            String[] split = line.split(",");
            String id = split[0];
            double price = Double.parseDouble(split[1]);
            return new Transaction(id, price);
        });

        //按照账户id分组
        KeyedStream<Transaction, String> keyByDS = transDS.keyBy(Transaction::getId);

        keyByDS.process(new KeyedProcessFunction<String, Transaction, Alert>() {

            //每一个task中共享同一个变量，同一个task中不能同key会冲突
            //在flink中需要通过状态来保存中间结果
            boolean flag = false;
            double headPrice = 0;

            //processElement 每一条数据执行一次
            @Override
            public void processElement(Transaction transaction,
                                       KeyedProcessFunction<String, Transaction, Alert>.Context ctx,
                                       Collector<Alert> out) throws Exception {

                String id = ctx.getCurrentKey();

                /*
                 * 于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
                 */
                //获取当前数据的交易金额
                Double price = transaction.getPrice();

                if (flag) {
                    //紧跟着一个大于 $500 的交易
                    if (price > 500) {
                        System.out.println("紧跟着一个大于 $500 的交易");
                        //发出告警信息
                        Alert alert = new Alert();
                        alert.setId(id);
                        alert.setHeadPrice(headPrice);
                        alert.setLasePrice(price);
                        //将数据发送到下游
                        out.collect(alert);
                    }
                    //重置flag
                    flag = false;
                }
                //如果出现小于 $1 美元的交易
                if (price < 1) {
                    System.out.println("如果出现小于 $1 美元的交易");
                    flag = true;
                    headPrice = price;
                }
            }
        }).print();

        env.execute();

        /*
         * 001,100
         * 001,200
         * 001,0.01
         * 001,600
         * 001,200
         * 001,0.2
         * 001,300
         * 001,700
         */


    }
}


